package com.arch.flume.interceptor;

import com.alibaba.fastjson2.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

/**
 * 自定义拦截器
 * @author pizhihui
 * @date 2017-11-11
 */
public class JsonInterceptor implements Interceptor{
    private String[] schema; //id,name,age,fv
    private String separator;

    public JsonInterceptor(String fields, String separator) {
        this.schema = fields.split("[,]");
        this.separator = separator;
    }

    @Override
    public void initialize() {
        // no-op
    }

    @Override
    public Event intercept(Event event) {
        Map<String, String> tuple = new LinkedHashMap<>();
        // 获取到源端采集到的数据
        String line = new String(event.getBody());
        String[] fields = line.split(separator);
        for(int i = 0; i < schema.length; i++) {
            String key = schema[i];
            String value = fields[i];
            tuple.put(key, value);
        }
        String res = JSONObject.toJSONString(tuple);
        event.setBody(res.getBytes());
        return null;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        for (Event event : events) {
            intercept(event);
        }
        return events;
    }

    @Override
    public void close() {
        // no-op
    }


    /**
     * Builder which builds new instances of the JsonInterceptor.
     */
    public static class Builder implements Interceptor.Builder {

        private String fields;
        private String separator;

        @Override
        public Interceptor build() {
            return new JsonInterceptor(fields, separator);
        }

        @Override
        public void configure(Context context) {
            // 需要添加的schema
            fields = context.getString("fields");
            // 数据的每行的分隔符
            separator = context.getString("separator");
        }
    }



}
